
# -*- coding: utf-8 -*-
from datetime import timedelta
from utils.operators.cluster_for_spark_sql_operator import SparkSqlOperator
from jms.ods.oms.yl_oms_oms_waybill_input_hi_check import jms_ods__yl_oms_oms_waybill_input_hi_check
# from utils.operators.spark_submit import SparkSubmitOperator
from airflow.models import Variable
from utils import combine
from utils.alerts import dingding_task_alert
from utils.alerts.phone_call import call_phone_by_email
from utils.alerts.yl_threeSegCodeOnFailue import yl_threeSegCodeOnFailure
from utils.alerts.yl_threeSegCodeOnSuccess import yl_threeSegCodeOnSuccess
def kwargs():
    kwargs = {
        "db": "jms_ods",
        "table": "yl_oms_oms_waybill",
        "desc": "ods层运单表",
        "taskid": "10070",
        "ifprivacy": 1,
        "warnignore": 0,
    }
    return kwargs
env = Variable.get('env', '')
jms_ods__yl_oms_oms_waybill = SparkSqlOperator(
    task_id='jms_ods__yl_oms_oms_waybill',
    email=['rongguangfan@jtexpress.com', 'yl_bigdata@yl-scm.com'],
    master='yarn',
    name='jms_ods__yl_oms_oms_waybill_{{ execution_date | date_add(1) | cst_ds }}',
    sql='jms/ods/oms/yl_oms_oms_waybill/execute.sql',
    pool_slots=2,
    driver_memory='2G',
    driver_cores=2,
    executor_cores=8,
    executor_memory='14G',
    num_executors=50,
    conf={'spark.dynamicAllocation.enabled'                  : 'true',
          'spark.shuffle.service.enabled'                    : 'true',
          'spark.dynamicAllocation.maxExecutors'             : 80,
          'spark.dynamicAllocation.cachedExecutorIdleTimeout': 30,
          'spark.sql.sources.partitionOverwriteMode'         : 'dynamic',
          'spark.executor.memoryOverhead'                    : '2G',
          'spark.sql.shuffle.partitions': 1400,
          'spark.default.paralleism': 1400,
          'spark.shuffle.consolidateFiles': 'true',
          'spark.shuffle.memoryFraction': '0.8',
          'spark.executor.extraJavaOptions': '-XX:+UseG1GC -XX:ParallelGCThreads=8',
          'spark.driver.extraJavaOptions': '-XX:+UseG1GC'
          },
    hiveconf={'hive.exec.dynamic.partition'             : 'true',  # 动态分区
              'hive.exec.dynamic.partition.mode'        : 'nonstrict',
              'hive.exec.max.dynamic.partitions'        : 300,  # 最大分区
              'hive.exec.max.dynamic.partitions.pernode': 300,  # 最大分区
              },
    yarn_queue='pro',
    email_on_retry=env.upper() == 'PRO',
    on_retry_callback=combine([dingding_task_alert(dingding_conn_id='dingding_bigdata'),
                               call_phone_by_email()],
                              ignore_error=True) if env.upper() == 'PRO' else None,
    execution_timeout=timedelta(minutes=30),
    on_success_callback=yl_threeSegCodeOnSuccess(kwargs(), dingding_conn_id="dingding_ThreeSeg_etl_info"),
    on_failure_callback=yl_threeSegCodeOnFailure(kwargs(), dingding_conn_id="dingding_ThreeSeg_etl_alert"),
)

jms_ods__yl_oms_oms_waybill << [jms_ods__yl_oms_oms_waybill_input_hi_check]
